﻿using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
namespace Athena.Transport.Implementations
{
    using ExceptionHandler;
    using Message;
    using Options;
    using Runtime.Server;
    public class TransportClient : ITransportClient, IDisposable
    {
        private readonly TransportOption _transportOption;
        private readonly ITransportMessageSender _transportMessageSender;
        private readonly ITransportMessageListener _transportMessageListener;
        private readonly IServiceExecutor _serviceExecutor;
        private readonly ILogger _logger;

        private readonly ConcurrentDictionary<string, TaskCompletionSource<RpcTransportMessage>> _rpcResultDictionary = new ConcurrentDictionary<string, TaskCompletionSource<RpcTransportMessage>>();

        public TransportClient(TransportOption transportOption, ITransportMessageSender transportMessageSender, ITransportMessageListener transportMessageListener, ILogger logger)
        {
            _transportOption = transportOption;
            _transportMessageSender = transportMessageSender;
            _transportMessageListener = transportMessageListener;
            _logger = logger;
            _transportMessageListener.Received += MessageListenerReveiced;
        }

        public async Task<RpcInvokeResultMessage> SendAsync(RpcInvokeMessage rpcInvokerMessage)
        {

            if (_logger.IsEnabled(LogLevel.Debug))
            {
                _logger.LogDebug($"准备发送消息 Id:{rpcInvokerMessage.CorrelationId}");
            }
            var rpcTransportMessage = RpcTransportMessage.CreateRpcInvokeMessage(rpcInvokerMessage);

            var invokeMessageCallback = RegisterResultCompletionSource(rpcTransportMessage, rpcInvokerMessage);

            try
            {
                await _transportMessageSender.SendAndFlushAsync(rpcTransportMessage);
            }
            catch (Exception exception)
            {
                _logger.LogError(exception, exception.Message);
                throw new RpcCommunicationException(exception.Message, exception);
            }
            return await invokeMessageCallback;
        }

        public void Dispose()
        {
            (_transportMessageSender as IDisposable)?.Dispose();
            (_transportMessageListener as IDisposable)?.Dispose();
            foreach (var item in _rpcResultDictionary.Values)
            {
                item.SetCanceled();
            }
        }



        private async Task<RpcInvokeResultMessage> RegisterResultCompletionSource(RpcTransportMessage rpcTransportMessage, RpcInvokeMessage rpcInvokeMessage)
        {

            if (_logger.IsEnabled(LogLevel.Debug))
            {
                _logger.LogDebug($"注册 Id:{rpcTransportMessage.Id}-{rpcInvokeMessage.CorrelationId} 响应事件");
            }
            var taskCompletionSource = new TaskCompletionSource<RpcTransportMessage>();

            _rpcResultDictionary.TryAdd(rpcTransportMessage.Id, taskCompletionSource);
            try
            {
                var cancellationTokenSource = new CancellationTokenSource(_transportOption.TimeoutMillseconds);
                using (cancellationTokenSource.Token.Register(() =>
                {
                    taskCompletionSource.SetException(new RpcException(AthenaRpcInvokeResultCodeDisplay.RpcTimeoutException, $"请求 Id:{rpcTransportMessage.Id}-{rpcInvokeMessage.CorrelationId} 响应超时。"));
                }))
                {
                    var result = await taskCompletionSource.Task;
                    return result.GetContent<RpcInvokeResultMessage>();
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
                throw ex;
            }
            finally
            {
                _rpcResultDictionary.TryRemove(rpcTransportMessage.Id, out TaskCompletionSource<RpcTransportMessage> value);
            }
        }

        private async Task MessageListenerReveiced(ITransportMessageSender transportMessageSender, RpcTransportMessage rpcTransportMessage)
        {
            if (_logger.IsEnabled(LogLevel.Debug))
            {
                _logger.LogInformation($"触发响应事件 Id:{rpcTransportMessage.Id}");
            }

            if (!_rpcResultDictionary.TryGetValue(rpcTransportMessage.Id, out TaskCompletionSource<RpcTransportMessage> taskCompletionSource))
            {
                return;
            }
            if (rpcTransportMessage.IsInvokeResultMessage)
            {
                var rpcInvokeResultMessage = rpcTransportMessage.GetContent<RpcInvokeResultMessage>();
                if (!rpcInvokeResultMessage.Code.Equals((int)AthenaRpcInvokeResultCodeDisplay.成功))
                {
                    taskCompletionSource.SetException(new RpcCommunicationException(rpcInvokeResultMessage.Message, null));
                }
                else
                {
                    taskCompletionSource.SetResult(rpcTransportMessage);
                }
            }

            if (rpcTransportMessage.IsInvokeMessage && _serviceExecutor != null)
            {
                await _serviceExecutor.ExecuteAsync(transportMessageSender, rpcTransportMessage);
            }
        }
    }
}
